package worker

import (
	"context"
	"distributed-task-scheduler/crontab/common"
	"time"

	"go.mongodb.org/mongo-driver/mongo/options"

	"go.mongodb.org/mongo-driver/mongo"
)

// mongodb存储日志
type LogSink struct {
	client         *mongo.Client
	logCollection  *mongo.Collection
	logChan        chan *common.JobLog
	autoCommitChan chan *common.LogBatch
}

var (
	// 单例
	G_logSink *LogSink
)

// 批量写入日志
func (logSink *LogSink) saveLogs(batch *common.LogBatch) {
	// 日志非必要，写入不成功也没关系
	logSink.logCollection.InsertMany(context.TODO(), batch.Logs)
}

// 日志存储协程
func (logSink *LogSink) writeLoop() {
	var (
		log          *common.JobLog
		logBatch     *common.LogBatch // 当前的批次
		commitTimer  *time.Timer
		timeoutBatch *common.LogBatch // 超时批次
	)
	for {
		select {
		case log = <-logSink.logChan:
			// 把这条log写到mongodb中  ->  logSink.logCollection.insertOne
			// 每次插入需要等待mongodb的一次请求往返，耗时可能因为网络慢花费比较长的时间
			// 故将日志攒一下，批次插入

			// 初始化批次
			if logBatch == nil {
				logBatch = &common.LogBatch{}
				// 让这个批次超时自动提交 (给1秒的时间) -> 数量较少的话，攒批次可能要攒很久，故定义一个时间超时提交
				commitTimer = time.AfterFunc(
					time.Duration(G_config.JobLogCommitTimeout)*time.Millisecond,
					// 闭包：func外面的logBatch经过append追加，地址可能会变，
					// 故应使用函数式生成函数，将批次使用参数传入
					func(batch *common.LogBatch) func() {
						return func() {
							// 发出超时通知，不用直接提交batch
							logSink.autoCommitChan <- batch
						}
					}(logBatch),
				)
			}

			// 把新日志追加到批次中
			logBatch.Logs = append(logBatch.Logs, log)

			// 如果批次满了，就立即发送
			if len(logBatch.Logs) >= G_config.JobLogBatchSize {
				// 发送日志
				logSink.saveLogs(logBatch)
				// 清空logBatch
				logBatch = nil
				// 取消定时器 -> 若是触发批次满了提交，就停止超时自动提交
				commitTimer.Stop()
			}

		case timeoutBatch = <-logSink.autoCommitChan: // 过期的批次
			// 判断过期批次是否仍旧是当前的批次 -> 批次满时，同时达到1秒超时时间，判断是不是已经被提交了，提交了就不再提交
			if timeoutBatch != logBatch {
				continue // 跳过已经被提交的批次
			}
			// 把批次写入到mongo中
			logSink.saveLogs(timeoutBatch)
			// 清空logBatch
			logBatch = nil
		}
	}
}

func InitLogSink() (err error) {
	var (
		client *mongo.Client
		opt    *options.ClientOptions
	)

	// 建立mongodb连接
	opt = options.Client()
	opt.ApplyURI(G_config.MongodbUri)
	opt.SetConnectTimeout(time.Duration(G_config.MongodbConnectTimeout) * time.Second)
	if client, err = mongo.Connect(context.TODO(), opt); err != nil {
		return
	}

	// 选择db和collection
	G_logSink = &LogSink{
		client:         client,
		logCollection:  client.Database("cron").Collection("log"),
		logChan:        make(chan *common.JobLog, 1000),
		autoCommitChan: make(chan *common.LogBatch, 1000),
	}

	// 启动一个mongodb处理协程
	go G_logSink.writeLoop()

	return
}

// 发送日志
func (logSink *LogSink) Append(jobLog *common.JobLog) {
	select {
	case logSink.logChan <- jobLog: // 若logChan满了，就会阻塞进入default
	default:
		// 队列满了就丢弃
	}
}
